package com.taoz.rpc.core.proxy;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.taoz.rpc.core.config.ApplicationConfig;
import com.taoz.rpc.core.fault.retry.RetryHelper;
import com.taoz.rpc.core.fault.retry.RetryHelperFactory;
import com.taoz.rpc.core.loadbalancer.LoadBalancer;
import com.taoz.rpc.core.loadbalancer.LoadBalancerFactory;
import com.taoz.rpc.core.model.RpcRequest;
import com.taoz.rpc.core.model.RpcResponse;
import com.taoz.rpc.core.model.ServiceMetaInfo;
import com.taoz.rpc.core.registry.Registry;
import com.taoz.rpc.core.registry.RegistryFactory;
import com.taoz.rpc.core.serializer.DefaultSerializer;
import com.taoz.rpc.core.serializer.Serializer;
import com.taoz.rpc.core.serializer.SerializerFactory;
import com.taoz.rpc.core.server.tcp.ProtocolConstant;
import com.taoz.rpc.core.server.tcp.ProtocolMessage;
import com.taoz.rpc.core.server.tcp.ProtocolStatus;
import com.taoz.rpc.core.server.tcp.ProtocolType;
import dev.failsafe.Call;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;
import io.vertx.core.parsetools.RecordParser;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

@Slf4j
public class ServiceProxyHandler implements InvocationHandler {

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Exception {
        RpcRequest rpcRequest = RpcRequest.builder()
                .serviceName(method.getDeclaringClass().getName())
                .methodName(method.getName())
                .parameterTypes(method.getParameterTypes())
                .args(args)
                .build();
        ApplicationConfig config = ApplicationConfig.getConfig();

        Registry registry = RegistryFactory.getRegistry(config.getRegistryName());
        ServiceMetaInfo info = new ServiceMetaInfo();
        info.setName(method.getDeclaringClass().getName());
        List<ServiceMetaInfo> serviceList = registry.serviceDiscovery(info);
        if (CollUtil.isEmpty(serviceList)) {
            log.info("未监测到有效服务");
            return null;
        }
        ServiceMetaInfo service = LoadBalancerFactory.getLoadBalancer().selectService(serviceList, null);
        log.info("准备访问服务：" + service.getServiceAddress());
        int port = service.getPort();
        String host = service.getHost();

        // 发送请求并返回得到的结果
        RetryHelper retryHelper = RetryHelperFactory.getRetryHelper();
        RpcResponse rpcResponse = retryHelper.getResultWithRetry(new Callable<RpcResponse>() {
            @Override
            public RpcResponse call() throws Exception {
                return doRequest(port, host, rpcRequest);
            }
        });

        return rpcResponse.getData();
    }

    RpcResponse doRequest(int port, String host, RpcRequest rpcRequest) throws ExecutionException, InterruptedException {
        Vertx vertx = Vertx.vertx();
        NetClient client = vertx.createNetClient();
        CompletableFuture<RpcResponse> future = new CompletableFuture<>();
        client.connect(port, host, conn -> {
            if (conn.succeeded()) {
                log.info("连接成功");
                NetSocket socket = conn.result();
                ProtocolMessage.Header header = new ProtocolMessage.Header();
                header.setMagic(ProtocolConstant.MAGIC);
                header.setVersion(ProtocolConstant.VERSION);
                header.setType(ProtocolType.REQUEST);
                header.setStatus(ProtocolStatus.OK);
                ProtocolMessage<RpcRequest> message = new ProtocolMessage<>(header, rpcRequest);
                socket.write(ProtocolMessage.toBuffer(message));

                RecordParser parser = RecordParser.newFixed(8);
                parser.setOutput(new Handler<Buffer>() {
                    int bodySize = -1;
                    Buffer resultBuffer = Buffer.buffer();
                    @Override
                    public void handle(Buffer buffer) {
                        resultBuffer.appendBuffer(buffer);
                        if (bodySize == -1) {
                            bodySize = buffer.getInt(4);
                            parser.fixedSizeMode(bodySize);
                        } else {
                            ProtocolMessage<?> resp = ProtocolMessage.toMessage(resultBuffer);
                            RpcResponse body = (RpcResponse) resp.getBody();
                            future.complete(body);
                            bodySize = -1;
                            resultBuffer = Buffer.buffer();
                        }
                    }
                });
                socket.handler(parser);
            } else {
                log.info("连接服务器失败");
                future.complete(null);
            }
        });
        RpcResponse rpcResponse = future.get();
        client.close();
        if (rpcResponse == null) {
            throw new RuntimeException("连接服务器失败");
        }
        return rpcResponse;
    }
}
